# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""common functions and classes used by some hook unittests."""

import logging
import os
import re
import sys
import unittest

import svn.core
import svn.fs

import gvn.userdb

import testing.testcommon

SUPER_GROUP = "super-group"

def CreateFile(fs, file_path, content="text", author="polly", logmsg="create",
                                                                    pool=None):
  """Adds a file to the test repository.
  """

  rev = svn.fs.youngest_rev(fs, pool)

  # Create transaction
  txn = svn.fs.begin_txn(fs, rev, pool)

  # Set basic revprops.
  svn.fs.change_txn_prop(txn, "svn:author", author, pool)
  svn.fs.change_txn_prop(txn, "svn:log", logmsg, pool)

  txnroot = svn.fs.txn_root(txn, pool)
  svn.fs.make_file(txnroot, file_path, pool)
  fp = svn.fs.apply_text(txnroot, file_path, None, pool)
  svn.core.svn_stream_write(fp, content)
  svn.core.svn_stream_close(fp)
  (conflicts, revision) = svn.fs.commit_txn(txn, pool)

def BuildPathCopyTxn(fs, path_from, txn_path, author, logmsg, pool, rev=None):
  """Create a test transaction from files in a directory path.

  Args:
    fs: svn_fs reference
    path_from: where on disk, to mirror a directory tree from
    txn_path: where in the depot to graft those files in, as a transation
    author: who owns that transaction
    logmsg: log message to link to the transaction
    pool: svn_pool reference
    rev: which revision this transaction is based on (None = HEAD)

  Returns:
    svn txn object

  This duplicates a file tree from path_from (without trailing slash)
  into a transaction inside txn_path (without a trailing slash)
  In many cases, txn_path should likely be empty to add those files at the
  top of the svn tree
  """

  assert (txn_path[-1] != "/" or len(txn_path) == 1), \
                    "trailing slash forbidden for txn_path in BuildPathCopyTxn"
  assert (path_from[-1] != "/" or len(path_from) == 1), \
                    "trailing slash forbidden in path_from BuildPathCopyTxn"

  if rev is None:
    rev = svn.fs.youngest_rev(fs, pool)

  # Create transaction
  txn = svn.fs.begin_txn(fs, rev, pool)

  # Set basic revprops.
  svn.fs.change_txn_prop(txn, "svn:author", author, pool)
  svn.fs.change_txn_prop(txn, "svn:log", logmsg, pool)

  txnroot = svn.fs.txn_root(txn, pool)

  def txn_mkdir(dir):
    try:
      svn.fs.make_dir(txnroot, dir, pool)
    except svn.core.SubversionException, e:
      if e.apr_err != svn.core.SVN_ERR_FS_ALREADY_EXISTS:
        raise

  # Create all the leading elements of the base path if needed
  def make_base_dirs(dir):
    if dir and dir !='/':
      make_base_dirs(os.path.dirname(dir))
    else:
      return
    txn_mkdir(dir)

  # first, recursively make the base directory where we graft our tree of files
  make_base_dirs(txn_path)
  for base_path, subdirs, files in os.walk(path_from):
    # non regex way to write base_path =~ s/^path_from/txn_path/ -> mk_txn_path
    mk_txn_path = '/'.join([txn_path, base_path[len(path_from)+1:]])
    txn_mkdir(mk_txn_path)
    for txn_file in files:
      txn_file_path = mk_txn_path + "/" + txn_file
      svn.fs.make_file(txnroot, txn_file_path, pool)
      fp = svn.fs.apply_text(txnroot, txn_file_path, None, pool)

      txn_file = open(base_path + "/" + txn_file)
      for line in txn_file:
        svn.core.svn_stream_write(fp, line)
      txn_file.close()
      svn.core.svn_stream_close(fp)

  # We've built our transaction, we can close/finalize it now that it's ready
  svn.fs.close_root(txnroot)
  return txn


def BuildPropTxn(fs, prop_name, prop_val, path, author, logmsg, pool, rev=None):
  """Create a test transaction with a file attribute on an existing path.

  Args:
    fs: svn_fs reference
    prop_name: which property to create
    prop_val: which property value to set
    path: which file or directory to attach the property to, or "REV" for
          rev property
    author: who owns that transaction
    logmsg: log message to link to the transaction
    pool: svn_pool reference
    rev: which revision this transaction is based on (None = HEAD)

  Returns:
    svn txn object

  This adds the attaches prop_name: prop_value to path.
  txn_path is required to point to an existing file or directory
  """

  assert (path == "REV" or path == "" or path[-1] != "/" or len(path) == 1), \
                            "trailing slash forbidden for path in BuildPropTxn"

  if rev is None:
    rev = svn.fs.youngest_rev(fs, pool)

  # Create transaction
  txn = svn.fs.begin_txn(fs, rev, pool)

  # Set basic revprops.
  svn.fs.change_txn_prop(txn, "svn:author", author, pool)
  svn.fs.change_txn_prop(txn, "svn:log", logmsg, pool)

  txnroot = svn.fs.txn_root(txn, pool)

  if path == "REV":
    svn.fs.change_txn_prop(txn, prop_name, prop_val, pool)
  else:
    svn.fs.change_node_prop(txnroot, path, prop_name, prop_val, pool)

  # We've built our transaction, we can close/finalize it now that it's ready
  svn.fs.close_root(txnroot)
  return txn


SVN_USERDB_PATH = "userdb"
class HookTestCase(unittest.TestCase):
  def setUp(self):
    """create temp repo, test group file, and logger object.
    """

    # open the rest repository stored in testdata/dumpfile
    self.repo_name = testing.testcommon.NewTmpRepository()
    # testcommon already cd's us in /tmp/tmpxxxxx under which we have
    # the testdata subdir

    # Setup logging object used by the hook
    # Hook debugging/tracing is done with export GVN_TEST_DEBUG=y before running
    if "GVN_TEST_DEBUG" in os.environ:
      level = logging.DEBUG
    else:
      level = logging.CRITICAL # This is equivalent to turning it off
    self._logger = logging.getLogger("test_check_owners")
    self._logger.setLevel(level)
    self._term_handler = logging.StreamHandler(sys.stderr)
    self._term_handler.setFormatter(logging.Formatter("%(message)s"))
    self._logger.addHandler(self._term_handler)

    self.userdb = gvn.userdb.UserDB(SVN_USERDB_PATH, create_db=True)


  def tearDown(self):
    """remove logging handler.
    """
    # apparently, the handler survives this object going out of scope, so you
    # have to manually remove it -- merlin
    # epg says this sucks :)
    self._logger.removeHandler(self._term_handler)

  def assertHookResult(self, actual_result, expected_result):
    """Looks for expected_result (regex|int) in actual_result.
    """
    self.assert_(type(actual_result) == type(expected_result),
                  "expected '%s' but got '%s'" % (expected_result,
                                                  actual_result))
    if isinstance(expected_result, int):
      self.assertEqual(actual_result, expected_result)
    else:
      if expected_result == actual_result:
        return True;
      else:
        self.assert_(re.search(expected_result, actual_result),
                     "expected '%s' but got '%s'" % (expected_result,
                                                     actual_result))


  # We provide 3 methods to test hooks:
  # _TestPaths allows to create a txn with files for a pre_commit hook
  # _TestProp does the same with a node prop or a revprop
  # _ChangeRevProps creates/changes/deleted a node or rev prop for a 
  #                 pre_revprop hook

  def _TestPaths(self, hook, local_path, repo_path, author, approve_user=False,
                 rev=4):
    """Return RunHook result triple from txn based on arguments.

    Args:
      local_path: where on disk, to mirror a directory tree from
      repo_path: where in the depot to graft those files in, as a transation
      author: who owns that transaction
      approve_user: optional user who is faked to have approved that revision
      rev: which revision the new txn is based on (default is 4)

    Returns:
      0 (pass), 1 (fail), string (fail and display string)

    Make a txn with data from local_path and check if author can submit it
    with its perms, or the perms from approve_user
    """
    # seed argv with expected arguments during a regular call, and overwrite
    # the ones we need to, with our own values lower down
    argv = ["pre-commit", "repo_name_bogus_ignored", "txnname_bogus_ignored"]
    hi = gvn.hooks.info.HookInfo(argv, userdb=self.userdb)
    hi._super_users = (set(), set([SUPER_GROUP]))

    # plug in the actual repo and txn objects, from which hi can compute fs
    hi._author = author
    hi._repo = svn.repos.open(self.repo_name, hi.pool)
    hi._txn = testing.hooks.common.BuildPathCopyTxn(hi.fs, local_path,
                              repo_path, author, "boguslog", hi.pool, rev)

    if approve_user:
      # Pretend this txn is a submit of a changebranch in rev
      # the hook doesn't ensure that's actually true.
      svn.fs.change_rev_prop(hi.fs, rev,
          "gvn:approve:%s" % approve_user, "", hi.pool)
      svn.fs.change_txn_prop(hi.txn, "gvn:change",
      # name is irrelevant except for rev field pointing to the approval
      # changelist
          "user1/boguschangenme@%d" % rev, hi.pool)

    ret = gvn.hooks.pre_commit.check_owners.RunHook(hi, self._logger)

    # Delete/clean up transaction
    svn.fs.abort_txn(hi.txn, hi.pool)
    return ret


  def _TestProp(self, hook, prop_name, prop_val, author="polly", path="",
                                                 approve_user=False, rev=None):
    """Runs pre_commit hook with txn built from given args and result triple.

    Args:
      hook: function pointer to hook that needs to be called
      prop_name: which property to create
      prop_val: which property value to set
      path: where in the depot to set the test property
            or pass "REV" for a revision property
      author: who owns that transaction
      approve_user: optional user who is faked to have approved that revision
      rev: If we want to base the txn off something else than the youngest

    Returns:
      0 (pass), 1 (fail), string (fail and display string)

    Make a txn by setting the node property to path (or making a revision prop
    if path is 'REV' and pass it to provided hook to check if author can
    submit it with its perms, or the perms from approve_user
    """
    # seed argv with expected arguments during a regular call, and overwrite
    # the ones we need to, with our own values lower down
    argv = ["pre-commit", "repo_name_bogus_ignored", "txnname_bogus_ignored"]
    hi = gvn.hooks.info.HookInfo(argv, userdb=self.userdb)

    # plug in the actual repo and txn objects, from which hi can compute fs
    hi._repo = svn.repos.open(self.repo_name, hi.pool)
    hi._txn = BuildPropTxn(hi.fs, prop_name, prop_val,
                                  path, author, "boguslog", hi.pool, rev=rev)

    if approve_user:
      # Pretend this txn is a submit of a changebranch in TESTREV
      # the hook doesn't ensure that's actually true.
      svn.fs.change_rev_prop(hi.fs, TESTREV,
          "gvn:approve:%s" % approve_user, "", hi.pool)
      svn.fs.change_txn_prop(hi.txn, "gvn:change",
      # name is irrelevant except for rev field pointing to the approval
      # changelist
          "user1/boguschangenme@%d" % TESTREV, hi.pool)

    hi._super_users = (set(), set([SUPER_GROUP]))

    ret = hook(hi, self._logger)

    # Delete/clean up transaction
    svn.fs.abort_txn(hi.txn, hi.pool)
    return ret

  def _ChangeRevProps(self, hook, user, prop_name, action, head=1,
                                                         rev=1, prop_value=""):
    """Runs pre_revprop hook with given arguments and return triple.

    Args:
      hook: function pointer to hook that needs to be called
      user: who wants to set/change/delete the property
      prop_name: which property to create
      action: A(dd)|D(elete)|M(odify)
      head: what's the head of our test repo (can't use youngest_rev w/o fs)
            pass a more accurate head if the unittest needs it
      rev: which version we're working on, if not 1
      prop_value: optional value for the property

    Returns:
      0 (pass), 1 (fail), string (fail and display string)

    Create a revprop change HookInfo and pass it to the hook for testing
    """

    # sanity check for tests
    self.assertTrue(action in ['A', 'M', 'D'])
    # seed argv with expected arguments during a regular call, and overwrite
    # the ones we need to, with our own values lower down
    argv = ["pre-revprop-change", "repo_name_bogus_ignored", rev,
                                                      user, prop_name, action]
    hi = gvn.hooks.info.HookInfo(argv, userdb=self.userdb)
    hi._head = head
    hi._super_users = SUPER_GROUP
    hi._input = prop_value
    # prevent hi.fs property from running in stubbed svn.fs.revision_prop(hi.fs
    # for test_check_gvn_revprops
    hi._fs = "foo"

    ret = hook(hi, self._logger)
    return ret
